package com.pz.gulimall.order.schedule;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.pz.gulimall.common.constant.MQConstant;
import com.pz.gulimall.common.to.mq.OrderReleaseTo;
import com.pz.gulimall.order.entity.MqMessageEntity;
import com.pz.gulimall.order.service.MqMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 定期搜索 mq_message 数据库，重发失败的消息
 */

@Slf4j
@Component
public class MsgSchedule {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private MqMessageService mqMessageService;

    // 处理错误抵达的消息（没有抵达mq服务器）
    @Scheduled(fixedRate = 1, timeUnit = TimeUnit.HOURS)
    public void reSendReachFailedMsg() {
        List<MqMessageEntity> reachFailedMessages = mqMessageService.list(new QueryWrapper<MqMessageEntity>().eq("message_status", MQConstant.MQStatus.REACH_FAILED.getCode()));

        reachFailedMessages.forEach(msg -> {
            try {
                if (msg.getClassType().equals("com.pz.gulimall.common.to.mq.OrderReleaseTo")) {
                    Class<?> aClass = Class.forName(msg.getClassType());
                    OrderReleaseTo orderReleaseTo = (OrderReleaseTo) JSON.to(aClass, msg.getContent());
                    CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
                    rabbitTemplate.convertAndSend(msg.getToExchange(),msg.getRoutingKey(), orderReleaseTo, cd);
                    cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            log.error("消息失败--{}",ex.getMessage());
                        }

                        @Override
                        public void onSuccess(CorrelationData.Confirm result) {
                            if(result.isAck()) {
                                mqMessageService.updateMsgStatus(msg.getMessageId(), MQConstant.MQStatus.REACHED.getCode());
                            }
                        }
                    });
                }
            } catch (Exception e) {
                log.info("异常重发消息-{}", e.getMessage());
            }
        });
    }
}
